package sputils

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/golog"
	"io"
	"runtime"
	"sync"
	"sync/atomic"
)

type SpBasePub struct {
	logger     golog.Logger
	lk         sync.RWMutex
	closedFlag int8

	pubSessionPrefix string
	pubsessionid     string
	pubconnstr       string
	stepMsg          string
	errmsg           string

	pubTemplate    string
	pubTopic       string
	pubTopicPrefix string
	pubTopicSuffix string

	conf      gobase.StrMap
	objectRef *gobase.ObjectInnerRef
	pubArg    interface{}
	subTopics map[string]struct{}
	sub       subpub.ISubA
	pub       subpub.IPub
}

var (
	pubnodesessionN int32
	basePubAliveN   int32
)

func GetBasePubAliveN() int32 {
	return atomic.LoadInt32(&basePubAliveN)
}

// subprefix 创建session时的前缀
func NewSpBasePub(subprefix string) *SpBasePub {
	rval := &SpBasePub{pubSessionPrefix: subprefix, logger: golog.Default()}
	atomic.AddInt32(&basePubAliveN, 1)
	runtime.SetFinalizer(rval, func(obj interface{}) {
		atomic.AddInt32(&basePubAliveN, -1)
	})
	return rval
}

func (this *SpBasePub) SetLogger(logger golog.Logger) {
	this.logger = logger
}

func (this *SpBasePub) GetSubSessionID() string {
	return this.pubsessionid
}

func (this *SpBasePub) ConfString() string {
	return this.pubconnstr
}

func (this *SpBasePub) GetDefaultTopic() string {
	return fmt.Sprintf("%s%s%s", this.pubTopicPrefix, this.pubTopic, this.pubTopicSuffix)
}

func (this *SpBasePub) GetPubTempl() string {
	return this.pubTemplate
}

func (this *SpBasePub) StatusText() string {
	this.lk.RLock()
	defer this.lk.RUnlock()
	objRef := this.objectRef
	if objRef == nil {
		return fmt.Sprintf("%s, step:%s", this.ConfString(), this.stepMsg)
	}
	s := objRef.GetStatus()
	if len(s) == 0 {
		return fmt.Sprintf("%s, step:%s", this.ConfString(), this.stepMsg)
	} else {
		return s
	}
}

func (this *SpBasePub) GetStatus(args ...interface{}) string {
	this.lk.RLock()
	defer this.lk.RUnlock()
	objRef := this.objectRef
	if objRef == nil {
		return fmt.Sprintf("%s, step:%s", this.ConfString(), this.stepMsg)
	}
	return objRef.GetStatus(args...)
}

func (this *SpBasePub) GetKeyValue(key string) (val interface{}, ok bool) {
	this.lk.RLock()
	defer this.lk.RUnlock()
	objRef := this.objectRef
	if objRef == nil {
		return
	}
	if lvIntf, ok1 := objRef.GetObject().(KeyValueGetter); ok1 {
		return lvIntf.GetKeyValue(key)
	}
	return
}

func (this *SpBasePub) closePub() {
	if this.objectRef != nil {
		this.stepMsg = fmt.Sprintf("[%s]closePub->ReleaseRef:0", gobase.NowString())
		sub := this.sub
		if sub != nil && len(this.subTopics) > 0 {
			for k, _ := range this.subTopics {
				sub.Unsub(this.pubsessionid, k)
			}
			this.subTopics = nil
		}

		this.objectRef.ReleaseRef()
		this.sub = nil
		this.pub = nil
		this.stepMsg = fmt.Sprintf("[%s]closePub->ReleaseRef:1", gobase.NowString())
		this.pubsessionid = ""
		this.pubconnstr = ""
		this.pubArg = nil
		this.objectRef = nil
	}
}

// 只停止当前， 不清理
func (this *SpBasePub) Stop() {
	this.lk.Lock()
	defer this.lk.Unlock()
	this.closePub()
}

// 清理不再使用
func (this *SpBasePub) Close() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	this.closedFlag = 1
	if len(this.pubsessionid) == 0 {
		return io.ErrClosedPipe
	}
	this.closePub()
	this.logger = nil
	return nil
}

func (this *SpBasePub) checkGetObj(conf gobase.StrMap) (*gobase.ObjectInnerRef, error) {
	conntype := conf.StringByName("conntype", "")
	conntype = fmt.Sprintf("sp.%s", conntype)
	obj, err := gobase.CreateFactoryInstance(conntype, conf)
	if err != nil {
		return nil, err
	}

	if lvintf, ok := obj.(*gobase.ObjectInnerRef); ok {
		return lvintf, nil
	} else {
		return nil, fmt.Errorf("%s:%s is not an object of [gobase.ObjectInnerRef]", conntype, ObjectType(obj))
	}
}

func (this *SpBasePub) Valid() bool {
	this.lk.RLock()
	defer this.lk.RUnlock()
	pub := this.pub
	if pub == nil {
		return false
	}

	return true
}

func (this *SpBasePub) HealthyCheck() (healthy gobase.Ternary, msg string) {
	this.lk.RLock()
	defer this.lk.RUnlock()
	if this.logger == nil {
		return gobase.TernaryFalse, "closed"
	}

	if this.pub == nil {
		return gobase.TernaryFalse, "pub is nil"
	}

	if this.objectRef == nil {
		return gobase.TernaryFalse, "objectRef is nil"
	}
	obj := this.objectRef.GetObject()
	if obj == nil {
		return gobase.TernaryFalse, "inner object is nil"
	}

	if lvIntf, ok := obj.(HealthyTernaryChecker); ok {
		return lvIntf.HealthyCheck()
	}

	if lvIntf, ok := obj.(HealthyInt8Checker); ok {
		v1, msg1 := lvIntf.HealthyCheck()
		return gobase.Ternary(v1), msg1
	}

	if lvIntf, ok := obj.(HealthyChecker); ok {
		v1, msg1 := lvIntf.HealthyCheck()
		if v1 {
			return gobase.TernaryTrue, msg1
		}
		return gobase.TernaryFalse, msg1
	}
	return
}

func (this *SpBasePub) ObjectFunc(fn func(obj interface{}) error) error {
	this.lk.RLock()
	defer this.lk.RUnlock()
	objRef := this.objectRef
	if objRef == nil {
		return gobase.ErrInvalidObject
	}
	return fn(objRef.GetObject())
}

func (this *SpBasePub) CheckPubJSONFunc(topic string, max int, fn func(json SuperValue)) {
	if this.Valid() {
		json := this.CreateTemplateJSON(false)
		fn(json)
		jsonBuf := json.AsJSONStringEx(false, false)
		this.Pub(topic, max, jsonBuf)
	}
}

func (this *SpBasePub) CheckPubFunc(fn func()) {
	if this.Valid() {
		fn()
	}
}

func (this *SpBasePub) Pub(topic string, max int, args ...interface{}) int {
	pub := this.pub
	if pub == nil {
		return -1
	}
	if len(topic) == 0 {
		topic = this.pubTopic
	}

	if len(this.pubTopicSuffix) > 0 && len(this.pubTopicSuffix) > 0 {
		topic = fmt.Sprintf("%s%s%s", this.pubTopicPrefix, topic, this.pubTopicSuffix)
	} else if len(this.pubTopicPrefix) > 0 {
		topic = fmt.Sprintf("%s%s", this.pubTopicPrefix, topic)
	} else if len(this.pubTopicSuffix) > 0 {
		topic = fmt.Sprintf("%s%s", topic, this.pubTopicSuffix)
	}

	if this.pubArg != nil {
		args = append(args, this.pubArg)
	}

	return pub.Pub(topic, max, args...)
}

func (this *SpBasePub) CreateTemplateJSON(cache bool) SuperValue {
	if defaultCreateSupperValueFunc == nil {
		panic("use SetDefaultCreateSupperValueFunc set default create func for suppervalue")
		return nil
	}
	if len(this.pubTemplate) == 0 {
		v, _ := defaultCreateSupperValueFunc(nil)
		return v
	}

	v, _ := defaultCreateSupperValueFunc([]byte(this.pubTemplate))
	return v
}

func (this *SpBasePub) checkPub(confMap gobase.StrMap, notify subpub.SubFunc) error {
	if len(this.pubsessionid) == 0 {
		this.pubsessionid = fmt.Sprintf("%s-pub-%d", this.pubSessionPrefix, atomic.AddInt32(&pubnodesessionN, 1))
	}
	this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:0", gobase.NowString())
	objref, err := this.checkGetObj(confMap)
	if err != nil {
		this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:-1, err:%s", gobase.NowString(), err.Error())
		this.logger.Warnf("create plug fail, args:%s, err:%s", confMap.URLEncode0(), err.Error())
		return err
	} else if objref == nil {
		this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:-1, err:%s", gobase.NowString(), "objref is null")
		this.logger.Warnf("create plug fail, args:%s", confMap.URLEncode0())
		return fmt.Errorf("create fail, args:%s", confMap.URLEncode0())
	} else {
		innerObj := objref.GetObject()
		if innerObj == nil {
			this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:-1, err:%s", gobase.NowString(), "innerObj is null")
			objref.ReleaseRef()
			this.logger.Warnf("create plug fail, inner object is nil, args:%s", confMap.URLEncode0())
			return fmt.Errorf("inner object is nil, args:%s", confMap.URLEncode0())
		}

		this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:1", gobase.NowString())
		this.objectRef = objref

		this.pubArg = ConvertArgs(objref.GetObject(), 0, confMap)

		if confMap.IntByName("evt", 1) == 1 {
			if notify != nil {
				if lvintf, ok := objref.GetObject().(subpub.ISubA); ok {
					subtopics := make(map[string]struct{})
					this.sub = lvintf
					this.sub.Sub(this.pubsessionid, SpEvtNotify, notify)
					this.sub.Sub(this.pubsessionid, SpEvtNotifyExt, notify)
					this.sub.Sub(this.pubsessionid, SpEvtExchange, notify)
					subtopics[SpEvtNotify] = struct{}{}
					subtopics[SpEvtNotifyExt] = struct{}{}
					subtopics[SpEvtExchange] = struct{}{}
				}
			}
		}
		this.pub = innerObj.(subpub.IPub)

		err = this.objectRef.CheckStart()
		if err == gobase.ErrUnsupported {
			this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:2, checkstart unsupoort!", gobase.NowString())
		} else {
			this.stepMsg = fmt.Sprintf("[%s]checkPubNode->checkGetObj:2", gobase.NowString())
		}

		return nil
	}
}

func (this *SpBasePub) ReloadConfString(str string, notify subpub.SubFunc) error {
	confMap := gobase.NewStrMap()
	confMap.URLFormDecode(str)
	return this.ReloadConfMap(confMap, notify)
}

func (this *SpBasePub) ReloadConfMap(confMap gobase.StrMap, notify subpub.SubFunc) error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if confMap == nil || len(confMap) == 0 {
		this.closePub()
		return nil
	}

	str := confMap.Encode("=", ";")
	if str == this.pubconnstr {
		return nil
	} else {
		this.closePub()
	}

	this.pubTopic = confMap["topic"]
	this.pubTopicPrefix = confMap["topicprefix"]
	this.pubTopicSuffix = confMap["topicsuffix"]
	this.pubTemplate = confMap["templ"]

	this.pubconnstr = str
	return this.checkPub(confMap, notify)
}

func (this *SpBasePub) ReloadConf(conf SuperValue, notifyfn subpub.SubFunc) error {
	confMap := gobase.NewStrMap()
	if !CheckIsNil(conf) {
		if conf.IsString() {
			confMap.URLFormDecode(conf.String())
		} else if conf.IsObject() {
			url := conf.StringByName("url", "")
			if len(url) > 0 {
				confMap.URLFormDecode(url)
			}
			conf.Write2StrMap(confMap)
		}
	}
	return this.ReloadConfMap(confMap, notifyfn)
}
